package cn.lyjuan.aliyun.mq.conf.java.conf.producer;

import cn.lyjuan.aliyun.mq.conf.java.callback.IgnoreSendCallback;
import cn.lyjuan.aliyun.mq.conf.java.conf.AMQUser;
import cn.lyjuan.base.util.JsonUtils;
import cn.lyjuan.base.util.StringUtils;
import com.aliyun.openservices.ons.api.*;
import com.aliyun.openservices.ons.api.bean.ProducerBean;

import java.nio.charset.Charset;
import java.util.Date;
import java.util.Properties;

/**
 * 适配器
 * Created by chad on 2016/11/9.
 */
public abstract class AMQProducer extends AMQUser implements Producer
{
    protected ProducerBean producer;

    @Override
    public void start()
    {
        prepare();// 前置任务

        Properties pro = new Properties();
        pro.setProperty(PropertyKeyConst.ProducerId, mqUserCst.getUserId());
        pro.setProperty(PropertyKeyConst.AccessKey, mqAuthCst.getAccessKey());
        pro.setProperty(PropertyKeyConst.SecretKey, mqAuthCst.getSecretKey());

        this.producer = new ProducerBean();
        producer.setProperties(pro);

        producer.start();
    }

    @Override
    public void shutdown()
    {
        producer.shutdown();
    }

    @Override
    public SendResult send(Message message)
    {
        return producer.send(message);
    }

    @Override
    public void sendOneway(Message message)
    {
        producer.sendOneway(message);
    }

    @Override
    public void sendAsync(Message message, SendCallback sendCallback)
    {
        producer.sendAsync(message, sendCallback);
    }

    @Override
    public boolean isStarted()
    {
        return producer.isStarted();
    }

    @Override
    public boolean isClosed()
    {
        return producer.isClosed();
    }

    /**
     * 发送JSON格式消息
     * @param key       全局唯一消息标识
     * @param jsonObj   转换为JSON格式的对象
     */
    public void sendMsgByJson(String key, Object jsonObj)
    {
        sendMsgByJson(key, jsonObj, null);
    }

    /**
     * 异步发送JSON格式消息
     * 忽略回调
     * @param key       全局唯一消息标识
     * @param jsonObj   转换为JSON格式的对象
     */
    public void sendMsgByJsonAsyncAndIgnoreCallback(String key, Object jsonObj)
    {
        byte[] body = JsonUtils.to(jsonObj).getBytes(Charset.forName(mqAuthCst.getCharset()));
        Message msg = pkgMsg(key, body);

        sendAsync(msg, new IgnoreSendCallback());
    }

    public void sendMsgByJson(String key, Object jsonObj, Date attime)
    {
        String body = JsonUtils.to(jsonObj);
        sendMsg(key, body, attime);
    }

    /**
     * 发送字符串消息
     * @param key       全局唯一消息标识
     * @param body      字符串消息
     */
    public void sendMsg(String key, String body)
    {
        this.sendMsg(key, body, null);
    }

    public void sendMsg(String key, String body, Date attime)
    {
        byte[] bodyBytes = body.getBytes(Charset.forName(mqAuthCst.getCharset()));

        sendMsg(key, bodyBytes, attime);
    }

    /**
     * 发送消息
     * @param key       全局唯一消息标识
     * @param body      二进制消息内容
     */
    public void sendMsg(String key, byte[] body)
    {
        // 同步发送消息，只要不抛异常就是成功
        this.sendMsg(key, body, null);
    }

    /**
     * 发送延时消息
     * @param key
     * @param body
     * @param delayMillisecond      延时毫秒
     */
    public void sendMsg(String key, byte[] body, long delayMillisecond)
    {
        sendMsg(key, body, new Date(System.currentTimeMillis() + delayMillisecond));
    }

    /**
     * 发送定时消息
     * @param key
     * @param body
     * @param attime        消息被消费的时间
     */
    public void sendMsg(String key, byte[] body, Date attime)
    {
        Message msg = pkgMsg(key, body);
        if (null != attime && attime.getTime() > System.currentTimeMillis())
            msg.setStartDeliverTime(attime.getTime());

        this.send(msg);
    }

    /**
     * 封装消息
     * @param key
     * @param body
     * @return
     */
    private Message pkgMsg(String key, byte[] body)
    {
        key = pkgKey(key, mqUserCst.getMsgTag());

        return new Message(mqUserCst.getTopicId(), mqUserCst.getMsgTag(), key, body);
    }

    /**
     * 封装消息KEY
     * @param key
     * @param msgTag
     * @return
     */
    private String pkgKey(String key, String msgTag)
    {
        if (!StringUtils.isNull(key) && !key.contains(msgTag))
            key = msgTag + "_" + key;

        return key;
    }
}
